package com.gzyj.flink.gps.map.nolocate;

import cn.hutool.core.lang.Pair;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.alibaba.fastjson.support.spring.FastJsonRedisSerializer;
import com.gzyj.flink.driver.Driverlogininfo;
import com.gzyj.flink.dto.alarm.Kafka_VehicleRemindContent;
import com.gzyj.flink.dto.alarm.NoLocateData;
import com.gzyj.flink.dto.alarm.RemindContent;
import com.gzyj.flink.gps.GPSPosition;
import com.gzyj.flink.gps.config.RedisKeyPrefixConfig;
import com.gzyj.flink.gps.factory.MySQLMybatisSessionFactory;
import com.gzyj.flink.gps.factory.RedisSessionFactory;
import com.gzyj.flink.gps.map.fatiguedriving.FatigueDrivingResult;
import com.gzyj.flink.gps.mapper.mysql.MonBudingweiSetMapper;
import com.gzyj.flink.gps.mapper.mysql.TsAlarmRuleMapper;
import com.gzyj.flink.gps.service.NormalAlarmService;
import com.gzyj.flink.gps.utils.IntUtils;
import com.gzyj.flink.gps.utils.TimeUtils;
import com.gzyj.flink.pojo.mysql.set.MonBudingweiSet;
import com.gzyj.flink.pojo.mysql.set.TsAlarmRule;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;

import java.util.Date;
import java.util.concurrent.TimeUnit;

public class NoLocateDataProcess extends RichMapFunction<GPSPosition, FatigueDrivingResult> {

    LettuceConnectionFactory redisFactory;

    SqlSessionFactory mysql;

    NormalAlarmService normalAlarmService;

    private RedisTemplate<String, Object> getRedisTemplate(LettuceConnectionFactory redisSessionFactory) {
        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(redisSessionFactory);
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(new FastJsonRedisSerializer<>(Object.class));
        redisTemplate.setHashKeySerializer(new StringRedisSerializer());
        redisTemplate.setHashValueSerializer(new FastJsonRedisSerializer<>(Object.class));
        redisTemplate.afterPropertiesSet();
        return redisTemplate;
    }


    @Override
    public void open(Configuration parameters) throws Exception {
        redisFactory = RedisSessionFactory.getRedisFactory();
        mysql = MySQLMybatisSessionFactory.getSqlSessionFactory();
        normalAlarmService = new NormalAlarmService();
    }


    public NoLocateData TryGetData_NLD(String vehId, RedisTemplate<String, Object> clusterRedisTemplate) {
        Object obj = clusterRedisTemplate.opsForValue().get(RedisKeyPrefixConfig.RedisKey_NoLocate + vehId);
        if (obj == null) return new NoLocateData();
        else {
            return JSON.parseObject(JSON.toJSONString(obj, SerializerFeature.DisableCircularReferenceDetect), NoLocateData.class);
        }
    }

    public Pair<Boolean, MonBudingweiSet> GetRuleSetting(GPSPosition vehicle, MonBudingweiSetMapper monBudingweiSetMapper, TsAlarmRuleMapper tsAlarmRuleMapper) {
        MonBudingweiSet setting = getVehicleForNoLocateParamsDic(vehicle.getVehId(), monBudingweiSetMapper);
        if (setting != null) {
            return new Pair<>(true, setting);
        } else {
            TsAlarmRule rule = tsAlarmRuleMapper.selectByRuleTypeBuDingWei(vehicle.getShiyongxingzhi());
            if (rule == null || rule.getRuleFlag() == 0) {
                return new Pair<>(false, null);
            } else {
                setting = getVehicleForNoLocateParamsDic(rule.getRuleNo(), monBudingweiSetMapper);
                if (setting != null) {
                    return new Pair<>(true, setting);
                } else {
                    return new Pair<>(false, null);
                }
            }


        }
    }


    private MonBudingweiSet getVehicleForNoLocateParamsDic(String vehId, MonBudingweiSetMapper monBudingweiSetMapper) {

        return monBudingweiSetMapper.selectByVehID(vehId);
    }

    public Driverlogininfo TryGetData_DLD(String vehId, RedisTemplate<String, Object> clusterRedisTemplate) {
        Object obj = clusterRedisTemplate.opsForValue().get(RedisKeyPrefixConfig.RedisKey_DriverSignOn + vehId);
        if (obj == null) {
            return new Driverlogininfo();
        } else {
            return JSON.parseObject(JSON.toJSONString(obj, SerializerFeature.DisableCircularReferenceDetect), Driverlogininfo.class);
        }

    }

    private RemindContent GenerateVehicleRemind(NoLocateData data, MonBudingweiSet setting, GPSPosition gps, String msgType) {

        if (msgType.equalsIgnoreCase("预警短信")) {
            RemindContent content = new RemindContent() {
                {
                    VehicleId = gps.getVehId();
                    Ipaddress = gps.getTerminalPhoneNo();
                    VehicleNo = gps.getCheliangpaizhao();
                    PlateColor = gps.getChepaiyanse();
                    RemindNote = setting.getTxnr();
                    RemindType = 0x01;
                    RemindTime = new Date();
                    MessageType = "不定位报警";
                    AlarmNote = "不定位报警";
                    AlarmGuid = data.SingleGuid;
                    GpsTime = data.GpsTime;
                    LastSpeed = gps.getSpeedExt();
                }

            };
            return content;
        } else if (msgType.equalsIgnoreCase("报警短信")) {
            RemindContent content = new RemindContent() {
                {
                    VehicleId = gps.getVehId();
                    Ipaddress = gps.getTerminalPhoneNo();
                    VehicleNo = gps.getCheliangpaizhao();
                    PlateColor = gps.getChepaiyanse();
                    RemindNote = setting.getTxnr();
                    RemindType = 0x02;
                    RemindTime = new Date();
                    MessageType = "不定位报警";
                    AlarmNote = "不定位报警";
                    AlarmGuid = data.SingleGuid;
                    GpsTime = data.GpsTime;
                    LastSpeed = gps.getSpeedExt();
                }

            };
            return content;
        } else {
            return null;
        }

    }

    private Kafka_VehicleRemindContent DisposeMessage(NoLocateData noLocateData, MonBudingweiSet mySetting, GPSPosition gps, String msgType) {
        if (msgType.equalsIgnoreCase("预警短信")) {
        } else if (msgType.equalsIgnoreCase("报警短信")) {
            if (mySetting.getTxcs() > 0 && StringUtils.isNotEmpty(mySetting.getTxnr())) {
                long ts = TimeUtils.getMinusSecond(new Date(), noLocateData.AlarmContentSendTime);
                if (ts >= mySetting.getTxjg() && noLocateData.AlarmContentCount < mySetting.getTxcs()) {
                    noLocateData.AlarmContentCount++;
                    noLocateData.AlarmContentSendTime = new Date();
                    RemindContent message = GenerateVehicleRemind(noLocateData, mySetting, gps, msgType);
                    return normalAlarmService.SendRemindToKafka(message);
                }
            }
        }
        return null;
    }

    @Override
    public FatigueDrivingResult map(GPSPosition gpsPosition) throws Exception {
        RedisTemplate<String, Object> clusterRedisTemplate = getRedisTemplate(redisFactory);
        SqlSession sqlSession = mysql.openSession(true);
        FatigueDrivingResult noLocateResult = new FatigueDrivingResult();
        try {

            MonBudingweiSetMapper monBudingweiSetMapper = sqlSession.getMapper(MonBudingweiSetMapper.class);
            TsAlarmRuleMapper tsAlarmRuleMapper = sqlSession.getMapper(TsAlarmRuleMapper.class);

            Driverlogininfo driverLoginData = TryGetData_DLD(gpsPosition.getVehId(), clusterRedisTemplate);

            NoLocateData noLocateData = TryGetData_NLD(gpsPosition.getVehId(), clusterRedisTemplate);
            if (noLocateData == null) return null;
            if (StringUtils.isEmpty(gpsPosition.getTIME())) {
                return null;
            }
            Pair<Boolean, Integer> p = TimeUtils.CheckData(noLocateData.getGpsTime(), gpsPosition.getTIME());
            int i = p.getValue();
            if (!p.getKey()) {
                return null;
            } else {
                noLocateData.LocalTime = new Date();
                noLocateData.GpsTime = gpsPosition.getGPSTimeExt();
            }

            //获取运算参数
            Pair<Boolean, MonBudingweiSet> p1 = GetRuleSetting(gpsPosition, monBudingweiSetMapper, tsAlarmRuleMapper);
            MonBudingweiSet mySetting = p1.getValue();
            if (p1.getKey()) {
                //如果车辆配置了规则
                if (mySetting.getEnable() == 1) {
                    int stat=IntUtils.LocalExt(gpsPosition.getSTATUSBIT());
                    if ( stat!= 1) {//终端没有开启定位
                        if (!noLocateData.IsCalc) {
                            noLocateData.ResetData();
                            noLocateData.StartCalc(gpsPosition);
                        }
                        noLocateData.WriteCalc(gpsPosition);
                        if (noLocateData.CalcConditionTimeSpan() > mySetting.getZxcxsj() * 60 && noLocateData.CalcConditionDistance() >=mySetting.getZxydjl()) {
                            if (!noLocateData.IsAlarm) {
                                noLocateData.StartAlarm(gpsPosition);
                                //从没报警切换到报警，发送开始报警
                                noLocateResult.setKafkaVehicleNormalAlarm(normalAlarmService.SendAlarmToKafka(noLocateData, gpsPosition, driverLoginData, true));
                            }
                            //其他的时候记录报警的过程数据
                            noLocateData.WriteAlarm(gpsPosition);
                            noLocateResult.setKafkaVehicleRemindContent(DisposeMessage(noLocateData, mySetting, gpsPosition, "报警短信"));
                        }
                    } else {

                        //定位了，停止报警
                        if (noLocateData.IsAlarm) {
                            noLocateData.StopAlarm(gpsPosition);
                            //从报警切换到没报警，发送停止报警
                            noLocateResult.setKafkaVehicleNormalAlarm(normalAlarmService.SendAlarmToKafka(noLocateData, gpsPosition, driverLoginData, false));
                        }
                        noLocateData.GenerateSingleGuid();
                        noLocateData.ResetData();

                    }

                } else {
                    //关闭了规则，停止报警
                    if (noLocateData.IsAlarm) {
                        noLocateData.StopAlarm(gpsPosition);
                        //从报警切换到没报警，发送停止报警
                        noLocateResult.setKafkaVehicleNormalAlarm(normalAlarmService.SendAlarmToKafka(noLocateData, gpsPosition, driverLoginData, false));
                    }
                    noLocateData.GenerateSingleGuid();
                    noLocateData.ResetData();
                }
            } else {

                //找不到规则，停止报警
                if (noLocateData.IsAlarm) {
                    noLocateData.StopAlarm(gpsPosition);
                    //从报警切换到没报警，发送停止报警
                    noLocateResult.setKafkaVehicleNormalAlarm(normalAlarmService.SendAlarmToKafka(noLocateData, gpsPosition, driverLoginData, false));
                }
                noLocateData.GenerateSingleGuid();
                noLocateData.ResetData();
            }

            clusterRedisTemplate.opsForValue().set(RedisKeyPrefixConfig.RedisKey_NoLocate + gpsPosition.getVehId(), noLocateData, 8, TimeUnit.HOURS);


        } catch (Exception e) {
            e.printStackTrace();
            sqlSession.rollback();
        } finally {
            sqlSession.close();
            return noLocateResult;

        }

    }
}
